package com.haiyou.common.thread.pool;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;
import com.haiyou.common.system.Properties;
import com.haiyou.common.thread.policy.DefaultRejectedPolicy;
import com.haiyou.common.thread.task.AbstractOrderedTask;
import com.haiyou.common.util.collect.CollectionUtils;
import com.haiyou.common.util.string.StringUtils;
import com.haiyou.common.util.system.SystemUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @Description: 线程安全线程池 ： 用户队列模型
 * @author xingyuan
 * @date 2022年2月15日 下午5:38:01
 */
@Slf4j
final public class UserQueuedThreadPoolExecutor implements Executor {

	final static AtomicLong tid = new AtomicLong();

	static long keepAliveTime;
	static TimeUnit keepAliveTimeUnit;
	static int taskCapacity;

	static {
		keepAliveTime = Properties.UserQueuedThreadPoolExecutor_keepAliveTime();
		keepAliveTimeUnit = Properties.UserQueuedThreadPoolExecutor_keepAliveTimeUnit();
		taskCapacity = Properties.UserQueuedThreadPoolExecutor_taskCapacity();
	}

	final String _threadNamedPrefix;

	final ThreadPoolExecutor _threadPoolExecutor;

	final Cache<Object, ConcurrentLinkedQueue<Runnable>> _userQueuedMap = Caffeine.newBuilder()
			.expireAfterAccess(keepAliveTime, keepAliveTimeUnit).build();

	final Cache<Object, AtomicBoolean> _userExecutingMap = Caffeine.newBuilder()
			.expireAfterAccess(keepAliveTime, keepAliveTimeUnit).build();

	public UserQueuedThreadPoolExecutor(int nThreads, String threadNamedPrefix) {
		this(nThreads, threadNamedPrefix, false);
	}

	/**
	 * 构造一个用户队列模型线程池
	 * 
	 * @param nThreads          线程池初始线程数，不会被销毁,最大线程数=nThreads*2
	 * @param threadNamedPrefix 线程命名前缀
	 */
	public UserQueuedThreadPoolExecutor(int nThreads, String threadNamedPrefix, boolean daemon) {
		this(nThreads, nThreads * 2, taskCapacity, true, threadNamedPrefix, daemon);
	}

	/**
	 * 构造一个用户队列模型线程池
	 * 
	 * @param corePoolSize      线程池初始线程数，不会被销毁
	 * @param maximumPoolSize   最大线程数，当队列满了时扩充，空闲后销毁
	 * @param taskCapacity      阻塞列队任务数量
	 * @param fair              是否使用公平锁
	 * @param threadNamedPrefix 线程命名前缀
	 */
	public UserQueuedThreadPoolExecutor(int corePoolSize, int maximumPoolSize, int taskCapacity, boolean fair,
			String threadNamedPrefix, boolean daemon) {
		_threadNamedPrefix = threadNamedPrefix;
		_threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, keepAliveTimeUnit,
				new ArrayBlockingQueue<Runnable>(taskCapacity, fair),
				(r) -> new Thread(r, StringUtils.merge(threadNamedPrefix, tid.getAndIncrement())) {
					{
						setDaemon(daemon);
					}
				}, DefaultRejectedPolicy.instance);
		log.warn("{} [{}] starting!!!", getClass().getSimpleName(), threadNamedPrefix);
	}

	private void shutdown() {
		_threadPoolExecutor.shutdown();
		log.warn("{} [{}] stopping!!!", getClass().getSimpleName(), _threadNamedPrefix);
	}

	private List<Runnable> shutdownNow() {
		return _threadPoolExecutor.shutdownNow();
	}

	private boolean isTerminated() {
		return _threadPoolExecutor.isTerminated();
	}

	public void shutdownNowGracefully() {
		List<Runnable> l = shutdownNow();
		if (!CollectionUtils.isEmpty(l)) {
			Queue<Runnable> q = Lists.newLinkedList(l);
			while (q.size() > 0) {
				Runnable r;
				try {
					r = q.poll();
					r.run();
				} catch (Exception e) {
					log.error(e.getMessage(), e);
				}
			}
		}
	}

	public void shutdownGracefully() {
		shutdown();
		while (!isTerminated())
			SystemUtils.sleep(1, TimeUnit.SECONDS);
		log.warn("{} [{}] stopped!!!", getClass().getSimpleName(), _threadNamedPrefix);
	}

	public <User> void addTask(User user, Runnable task) {
		if (_userQueuedMap.get(user, k -> Queues.newConcurrentLinkedQueue()).offer(task)) {
			final AtomicBoolean executing = _userExecutingMap.get(user, k -> new AtomicBoolean());
			if (executing.compareAndSet(false, true)) {
				_threadPoolExecutor.execute(() -> {
					ConcurrentLinkedQueue<Runnable> q = _userQueuedMap.getIfPresent(user);
					if (q != null) {
						while (!q.isEmpty()) {
							Runnable r = null;
							try {
								r = q.poll();
								r.run();
							} catch (Exception e) {
								log.error(e.getMessage() + ":" + r, e);
							} finally {
								if (q.isEmpty()) {
									executing.compareAndSet(executing.get(), false);
								}
							}
						}
					}
				});
			}
		}
	}

	@Override
	public void execute(Runnable command) {
		if (command instanceof AbstractOrderedTask) {
			addTask(((AbstractOrderedTask<?>) command).getSer(), command);
		} else {
			addTask(command, command);
		}
	}

}
